package com.atguigu.gmall.cdc.map;

import com.atguigu.gmall.cdc.vo.StudentVO;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import static com.atguigu.gmall.cdc.utils.CommonUtil.parseMysqlCDCData;
import static com.atguigu.gmall.cdc.utils.CommonUtil.parseStrToStuVO;

/**
 * @author: liminghui
 * @date: 2021/10/29 23:58
 * @version: 1.0
 * @description:
 */
public class MyMapFunction implements MapFunction<String, Tuple2<StudentVO, StudentVO>> {

    @Override
    public Tuple2<StudentVO, StudentVO> map(String input) throws Exception {

        Tuple2<StudentVO, StudentVO> tuple2 = null;

        // group是针对括号（）来说的，group（0）就是指的整个串，group（1） 指的是第一个括号里的东西，group（2）指的第二个括号里的东西。
        String beforeRegex = "before=Struct\\{?(\\S+?)\\}";
        String afterRegex = "after=Struct\\{?(\\S+?)\\}";

        String before = parseMysqlCDCData(input, beforeRegex);
        String after = parseMysqlCDCData(input, afterRegex);
        if (StringUtils.isEmpty(before) && StringUtils.isEmpty(after)) {
            tuple2 = null;
        } else if (!StringUtils.isEmpty(before) && !StringUtils.isEmpty(after)) { //更新
            StudentVO beforeStu = parseStrToStuVO(before);
            StudentVO afterStu = parseStrToStuVO(after);
            tuple2 = new Tuple2<>(beforeStu, afterStu);
        } else if (StringUtils.isEmpty(before) && !StringUtils.isEmpty(after)) { // 新增
            StudentVO afterStu = parseStrToStuVO(after);
            tuple2 = new Tuple2<>(null, afterStu);
        } else if (!StringUtils.isEmpty(before) && StringUtils.isEmpty(after)) { // 删除
            StudentVO beforeStu = parseStrToStuVO(before);
            tuple2 = new Tuple2<>(beforeStu, null);
        }

        return tuple2;
    }

}

